package com.tn.jinq.coherence;

import java.util.Iterator;
import java.util.Map;
import java.util.HashMap;
import java.util.Collection;
import java.util.Set;

import com.tangosol.util.Filter;
import com.tangosol.util.MapListener;
import com.tangosol.util.MapEvent;
import com.tangosol.net.NamedCache;
import com.tangosol.net.cache.ContinuousQueryCache;

import com.tn.jinq.DeleteException;
import com.tn.jinq.KeyedDeletable;
import com.tn.jinq.event.ChangeEventType;
import com.tn.jinq.event.KeyedChangeEvent;
import com.tn.jinq.predicate.Predicate;
import com.tn.jinq.Context;
import com.tn.jinq.Keyed;
import com.tn.jinq.Queryable;
import com.tn.jinq.KeyedWritable;
import com.tn.jinq.Observable;
import com.tn.jinq.FilteredObservable;
import com.tn.jinq.Observer;
import com.tn.jinq.DataGetException;
import com.tn.jinq.WriteException;
import com.tn.jinq.coherence.predicate.CoherencePredicateInterpreter;

/**
 * An implementation of <code>Context</code> that works with the values (as opposed to the keys) in a Coherence cache.
 */
public class CoherenceValueContext<K, V, P extends Predicate> extends Context<Filter>
  implements Keyed<K, V>, Queryable<V, P>, KeyedWritable<K, V>, KeyedDeletable<K, V>,
    Observable<KeyedChangeEvent<K, V>>, FilteredObservable<KeyedChangeEvent<K, V>, P>
{
  private NamedCache cache;
  private Map<Observer, ContinuousQueryCache> continuousQueryCaches;
  private Map<Observer, MapListener> listeners;
  private Map<Observer, Filter> observerFilters;

  /**
   * Creates a new <code>CoherenceValueContext</code>.
   *
   * @param cache the cache to wrap.
   */
  public CoherenceValueContext(NamedCache cache)
  {
    super(new CoherencePredicateInterpreter());
    this.cache = cache;
    this.continuousQueryCaches = new HashMap<Observer, ContinuousQueryCache>();
    this.listeners = new HashMap<Observer, MapListener>();
    this.observerFilters = new HashMap<Observer, Filter>();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public V get(K key) throws DataGetException
  {
    //noinspection unchecked
    return (V)cache.get(key);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Iterable<V> get(Collection<K> keys) throws DataGetException
  {
    //noinspection unchecked
    return cache.getAll(keys).values();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Set<K> keySet()
  {
    //noinspection unchecked
    return cache.keySet();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Iterable<V> select(final P predicate) throws DataGetException
  {
    return new Iterable<V>()
    {
      public Iterator<V> iterator()
      {
        //noinspection unchecked
        return new ValueIterator(cache.entrySet(getPredicateInterpreter().interpret(predicate)).iterator());
      }
    };
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void write(K key, V value) throws WriteException
  {
    //If there are continuousQueryCaches then check to see if the value can be added to one of the ContinuousQueryCaches
    //first.  The value only needs adding to one ContinuousQueryCaches for all MapListeners to be invoked.
    for (Observer observer : continuousQueryCaches.keySet())
    {
      Filter observerFilter = observerFilters.get(observer);
      if (observerFilter != null && observerFilter.evaluate(value))
      {
        ContinuousQueryCache cache = continuousQueryCaches.get(observer);
        cache.put(key, value);
        return;
      }
    }

    //noinspection unchecked
    this.cache.put(key, value);
  }

  /**
   * {@inheritDoc}
   */
  @SuppressWarnings("unchecked")
  @Override
  public V delete(K key) throws DeleteException
  {
    //If there are continuousQueryCaches then check to see if the value can be removed from one of the 
    //ContinuousQueryCaches first.  The value only needs adding to one ContinuousQueryCaches for all MapListeners to be 
    //invoked.
    for (Observer observer : continuousQueryCaches.keySet())
    {
      V value = (V)continuousQueryCaches.get(observer).get(key);
      
      Filter observerFilter = observerFilters.get(observer);
      if (observerFilter != null && observerFilter.evaluate(value))
      {
        ContinuousQueryCache cache = continuousQueryCaches.get(observer);
        return (V)cache.remove(key);
      }
    }

    return (V)this.cache.remove(key);
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void subscribe(final Observer<KeyedChangeEvent<K, V>> observer)
  {
    MapListener mapListener = new ObserverMapListener(observer);
    listeners.put(observer, mapListener);
    cache.addMapListener(mapListener);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void subscribe(final Observer<KeyedChangeEvent<K, V>> observer, P predicate)
  {
    if (!continuousQueryCaches.containsKey(observer))
    {
      Filter filter = getPredicateInterpreter().interpret(predicate);
      observerFilters.put(observer, filter);

      continuousQueryCaches.put(
        observer,
        new ContinuousQueryCache(
          cache,
          filter,
          new ObserverMapListener(observer)
        )
      );
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void unsubscribe(Observer<KeyedChangeEvent<K, V>> observer)
  {
    if (continuousQueryCaches.containsKey(observer))
    {
      continuousQueryCaches.remove(observer);
      observerFilters.remove(observer);
    }
    else if (listeners.containsKey(observer))
    {
      cache.removeMapListener(listeners.get(observer));
    }
  }

  /**
   * An implementation of <code>MapListener</code> that invokes an <code>Observer</code>.
   */
  private class ObserverMapListener implements MapListener
  {
    private Observer<KeyedChangeEvent<K, V>> observer;

    /**
     * Creates a new <code>ObserverMapListener</code> initialized with the <code>observer</code>.
     */
    private ObserverMapListener(Observer<KeyedChangeEvent<K, V>> observer)
    {
      this.observer = observer;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryInserted(MapEvent mapEvent)
    {
      //noinspection unchecked
      observer.onNext(new KeyedChangeEvent(ChangeEventType.INSERT, mapEvent.getKey(), mapEvent.getNewValue()));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryUpdated(MapEvent mapEvent)
    {
      //noinspection unchecked
      observer.onNext(new KeyedChangeEvent(ChangeEventType.UPDATE, mapEvent.getKey(), mapEvent.getNewValue()));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryDeleted(MapEvent mapEvent)
    {
      //noinspection unchecked
      observer.onNext(new KeyedChangeEvent(ChangeEventType.DELETE, mapEvent.getKey(), mapEvent.getNewValue()));
    }
  }

  /**
   * An implementation of <code>Iterator</code> that returns the values from a <code>Map.Entry</code>.
   */
  private class ValueIterator implements Iterator<V>
  {
    private Iterator<Map.Entry> entryIterator;

    /**
     * Creates a new <code>ValueIterator</code> initialized with the <code>entryIterator</code>.
     */
    private ValueIterator(Iterator<Map.Entry> entryIterator)
    {
      this.entryIterator = entryIterator;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean hasNext()
    {
      return entryIterator.hasNext();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public V next()
    {
      //noinspection unchecked
      return (V)entryIterator.next().getValue();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void remove()
    {
      entryIterator.remove();
    }
  }
}
